package com.maxmlr.concurrent.cyclic;

import java.util.*;
import java.util.concurrent.*;

/**
 * @ClassName TestCyclic1
 * @Description TODO
 * @Author maxiao
 * @Date 2020/4/11 下午10:21
 * @Version V1.0
 */
public class TestCyclic1 {
    private static Map<String, CyclicBarrier> barrierMap = new ConcurrentHashMap<>();
    private static Map<String, List<Message>> msgMap = new ConcurrentHashMap<>();
    private static ExecutorService s = Executors.newFixedThreadPool(5);

    public static void main(String[] args) {
        String key = "aaa";
        CyclicBarrier barrier = new CyclicBarrier(3, () -> {

            List<Message> messages = msgMap.get(key);
            long count = messages.stream().filter(Message::flag).count();
            if (count < 3) {
                System.out.println("main---restart");
                send(key, "000");
            } else {
                System.out.println("main---start");
            }

        });
        barrierMap.put(key, barrier);

        send(key, "001");

//        s.execute(new ConsumerInfo(key, "000", "111"));
    }

    private static void send(String key, String code) {
        String code1 = "111";
        if ("000".equals(code)) {
            code1 = "000";
        }
        code = "000";
        s.execute(new ProducerInfo(key, code1, "444"));
        s.execute(new ProducerInfo(key, code, "555"));
        s.execute(new ProducerInfo(key, code, "666"));
    }

    private static class ProducerInfo implements Runnable {
        private String key;
        private String code;
        private String value;

        public ProducerInfo(String key, String code, String value) {
            this.key = key;
            this.code = code;
            this.value = value;
        }

        @Override
        public void run() {

            System.out.println(Thread.currentThread().getName() + "--执行作务中。。。。:" + "key:" + key + ",code:" + code + ",value:" + value);

            /*try {
                //Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }*/

            s.execute(new ConsumerInfo(key, code, value));

        }

    }

    private static class ConsumerInfo implements Runnable {
        private Message msg;
        Random random = new Random();

        public ConsumerInfo(Message msg) {
            this.msg = msg;
        }

        public ConsumerInfo(String key, String code, String value) {
            this.msg = new Message();
            this.msg.key = key;
            this.msg.code = code;
            this.msg.value = value;
        }

        @Override
        public void run() {
            CyclicBarrier cyclicBarrier = barrierMap.get(msg.key);
            long l = System.currentTimeMillis();
            try {
                Thread.sleep(random.nextInt(1000));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            long l1 = System.currentTimeMillis() - l;
            System.out.println(Thread.currentThread().getName() + "==" + l1 + "--任务执行完成。。。。:" + msg);
            String key = msg.key;
            List<Message> messages = msgMap.get(key);
            if (Objects.isNull(messages)) {
                messages = new ArrayList<>();
            }
            messages.add(msg);
            msgMap.put(key, messages);
            try {
                int await = cyclicBarrier.await(500, TimeUnit.MILLISECONDS);
            } catch (Exception e) {
                System.err.println(Thread.currentThread().getName() + "----" + e.getMessage()
                        + "==" + cyclicBarrier.getNumberWaiting());
            }


        }

    }

}
